package com.gitee.wsl.coroutines.executor.sample

import com.gitee.wsl.concurrent.api.SuspendRunnableWithResult
import com.gitee.wsl.coroutines.executor.KExecutor
import com.gitee.wsl.coroutines.executor.SuspendKExecutor
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.async
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.coroutines.CoroutineContext


class AsyncExecutor : SuspendKExecutor {
    private val lock= Mutex()
    private var lastPromise: Deferred<*> = CompletableDeferred(Unit)

    suspend fun await() {
        while (true) {
            val cpromise = lock.withLock { lastPromise }
            cpromise.await()
            if (lock.withLock { cpromise == lastPromise }) break
        }
    }

    suspend fun cancel() = apply {
        lock.withLock { lastPromise }.cancel()
        lock.withLock { lastPromise = CompletableDeferred(Unit) }
    }

    private suspend operator fun <T> invoke(context: CoroutineContext, func: suspend () -> T): Deferred<T> = lock.withLock {
        val oldPromise = lastPromise
        CoroutineScope(context).async {
            oldPromise.await()
            func()
        }.also { lastPromise = it }
    }

    override suspend fun <T> executeSuspend(block: SuspendRunnableWithResult<T>): T {
        val task = invoke(currentCoroutineContext()){ block.run() }
        return task.await()
    }

    companion object{
        val INSTANCE = AsyncExecutor()
    }

}

fun KExecutor.Companion.asyncExecutor() = AsyncExecutor.INSTANCE


/**
 * Prevents a named invoke to happen at the same time (by effectively enqueuing by name).
 * This class is thread-safe.
 */
class NamedAsyncExecutor(val threadFactory: () -> SuspendKExecutor = { AsyncExecutor() }) {
    class AsyncJob(val thread: SuspendKExecutor) {
        var count = 0
    }
    //private val lock = NonRecursiveLock()
    private val lock= Mutex()

    private val jobs = LinkedHashMap<String, AsyncJob>()

    internal fun threadsCount() = jobs.size

    suspend operator fun <T> invoke(name: String, func: SuspendRunnableWithResult<T>): T {
        val job = lock.withLock {
            jobs.getOrPut(name) { AsyncJob(threadFactory()) }.also { it.count++ }
        }
        try {
            return job.thread.executeSuspend(func)
        } finally {
            // Synchronization to prevent another thread from being added in the mean time, or a process queued.
            lock.withLock {
                job.count--
                if (job.count == 0) {
                    jobs.remove(name)
                }
            }
        }
    }
}

fun KExecutor.Companion.namedAsyncExecutor(threadFactory: () -> SuspendKExecutor = { AsyncExecutor() }) = NamedAsyncExecutor(threadFactory)
